package day05

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 2.6	RDD序列化
 *
 * 在实际开发中我们往往需要自己定义一些对于RDD的操作，那么此时需要注意
 * 的是，初始化工作是在Driver端进行的，而实际运行程序是在Executor端
 * 进行的，这就涉及到了跨进程通信，是需要序列化的。
 */
object Serialization {

}

/**
 * 2.6.1	闭包检查
 */
object Serialization_closure_check {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.创建两个对象
    val user1: User = new User()
    user1.name = "zhangsan"

    val user2: User = new User()
    user2.name = "lisi"

    val userRDD1: RDD[User] = sc.makeRDD(List(user1, user2))

    //3.1 打印，ERROR报java.io.NotSerializableException
    userRDD1.foreach(user => println(user.name))


    //3.2 打印，RIGHT
    val userRDD2: RDD[User] = sc.makeRDD(List())
    //userRDD2.foreach(user => println(user.name))

    //3.3 打印，ERROR Task not serializable 注意：没执行就报错了
    userRDD2.foreach(user => println(user1.name))



    // 关闭连接
    sc.stop()
  }
}

// 未序列化
//class User {
//  var name: String = _
//}

// 序列化
class User extends Serializable {
  var name: String = _
}

/**
 * 2.6.2	序列化方法和属性
 *
 * 1）说明
 * Driver：算子以外的代码都是在Driver端执行
 * Executor：算子里面的代码都是在Executor端执行
 */
object Serialization_closure_method {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.创建一个RDD
    val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello spark", "hive", "atguigu"))

    val search: Search = new Search("hello")

    search.getMatch1(rdd).collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

class Search(query: String) {
  def isMatch(s: String): Boolean = {
    s.contains(query)
  }

  // 函数序列化案例
  def getMatch1(rdd: RDD[String]): RDD[String] = {
    //rdd.filter(this.isMatch)
    rdd.filter(isMatch)
  }

  // 属性序列化案例
  def getMatch2(rdd: RDD[String]): RDD[String] = {

    // rdd.filter(x=>x.contains(this.query))
    rdd.filter(x => x.contains(query))
    //    val q = query
    //    rdd.filter(x => x.contains(q))
  }
}

/**
 * 2.6.3	Kryo序列化框架
 *
 * 参考地址: https://github.com/EsotericSoftware/kryo
 *
 * Java的序列化能够序列化任何的类。但是比较重，序列化后对象的提交也比较大；
 *
 * Spark出于性能的考虑，Spark2.0开始支持另外一种Kryo序列化机制。
 * Kryo速度是Serializable的10倍。
 * 当RDD在Shuffle数据的时候，简单数据类型、数组和字符串类型已经在Spark内部使用kryo来序列化。
 *
 * 注意：即使使用kryo序列化，也要继承Serializable接口。
 */
object Serialization_framework_kryo {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf()
      .setAppName("SparkCoreTest")
      .setMaster("local[*]")
      // 替换默认的序列化机制
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      // 注册需要使用kryo序列化的自定义类
      .registerKryoClasses(Array(classOf[Searcher]))

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello atguigu", "atguigu", "hahah"), 2)

    val searcher: Searcher = Searcher("hello")
    val result: RDD[String] = searcher.getMatchedRDD1(rdd)
    result.collect.foreach(println)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 样例类(case class)是默认序列化的
 *
 * @param query
 */
case class Searcher(val query: String) {

  def isMatch(s: String): Boolean = {
    s.contains(query)
  }

  def getMatchedRDD1(rdd: RDD[String]): RDD[String] = {
    rdd.filter(isMatch)
  }

  def getMatchedRDD2(rdd: RDD[String]): RDD[String] = {
    val q: String = query
    rdd.filter(_.contains(q))
  }
}

